package com.lg.delay.queue.queue;

import com.alibaba.fastjson.JSON;
import com.lg.delay.queue.constant.RedisQueueKey;
import com.lg.delay.queue.constant.RetryStrategyEnum;
import com.lg.delay.queue.pojo.Job;
import com.lg.delay.queue.service.ConsumerService;
import com.lg.delay.queue.task.TaskManager;
import com.lg.delay.queue.utils.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

/**
 * Ready Queue 存放处于Ready状态的Job（并且是处于List状态下的，值存放Job Id）
 *
 * @author by Mr. Li 2020/12/9 17:32
 */
@Slf4j
@Component
public class ReadyQueueContext {

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private ConsumerService consumerService;

    /**
     * Topic 消费线程
     * 容器加载完成，调用之前 自动调用该方法
     */
    @PostConstruct
    public void startTopicConsumer() {
        TaskManager.doTask(this::runTopicThreads, "开启TOPIC消费线程");
    }

    private void runTopicThreads() {
        log.info("消费TOPIC线程启动");
        while (true) {
            RLock lock = null;
            try {
                log.info("尝试获取消费TOPIC线程分布式锁");
                lock = redissonClient.getLock(RedisQueueKey.CONSUMER_TOPIC_LOCK);

            } catch (Exception e) {
                log.error("runTopicThreads getLock error", e);
            }
            try {
                // 获取锁失败
                if (lock == null) {
                    continue;
                }
                // 分布式锁时间比Blpop阻塞时间多1S，避免出现释放锁的时候，锁已经超时释放，unlock报错
                boolean b = lock.tryLock(RedisQueueKey.LOCK_WAIT_TIME, RedisQueueKey.LOCK_RELEASE_TIME, TimeUnit.SECONDS);
                if (!b) {
                    continue;
                }

                // 获取ReadyQueue中待消费的数据
                RBlockingDeque<String> queue = redissonClient.getBlockingDeque(RedisQueueKey.RD_LIST_TOPIC_PRE);
                String topicId = queue.poll(60, TimeUnit.SECONDS);
                if (StringUtils.isEmpty(topicId)) {
                    continue;
                }

                // 获取job元信息内容
                RMap<Object, Job> jobPoolMap = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY);
                Job job = jobPoolMap.get(topicId);

                // 消费获取到的job
                FutureTask<Boolean> taskResult = TaskManager.doFutureTask(() -> consumerService.consumerMessage(job.getUrl(), job.getBody()),
                        job.getTopic() + "-->消费JobId-->" + job.getJobId());
                if (taskResult.get()) { // 其实此时这样处理是不当的，再调用get之后，线程会被阻塞住。这里更加客观的方式，是将消费的逻辑扔到消息队列中由空闲线程进行消费会比较妥当
                    jobPoolMap.remove(topicId);// 消费成功
                } else {
                    int retrySum = job.getRetry() + 1;// 记录重试次数
                    // 重试次数不能大于5
                    if (retrySum > RetryStrategyEnum.RETRY_FIVE.getRetry()) {
                        jobPoolMap.remove(topicId);
                        continue;
                    }
                    job.setRetry(retrySum);
                    long nextTime = job.getDelay() + RetryStrategyEnum.getDelayTime(job.getRetry()) * 1000;
                    log.info("next retryTime is [{}]", DateUtil.long2Str(nextTime));
                    RScoredSortedSet<Object> delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
                    delayBucket.add(nextTime, topicId);
                    // 更新元信息失败次数
                    jobPoolMap.put(topicId, job);
                }

            } catch (InterruptedException | ExecutionException e) {
                log.error("runTopicThreads error", e);
            } finally {
                if (lock != null) {
                    try {
                        lock.unlock();
                    } catch (Exception e) {
                        log.error("runTopicThreads error", e);
                    }
                }
            }

        }
    }

    public static void main(String[] args) {
        Job job = new Job();
        job.setJobId(UUID.randomUUID().toString());
        job.setTopic("NOTIFY");
        long delay = System.currentTimeMillis() + 10000;
        DateUtil.long2Str(delay);
        job.setDelay(delay);
        job.setBody("{\"amount\":0.01}");
        job.setUrl("http://localhost:8080/testCallBack");
        System.out.println(JSON.toJSONString(job));
    }

}
